package com.bbx.flink.table_api_or_sql;

import com.bbx.flink.demo.entity.Student;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;

import static org.apache.flink.table.api.Expressions.$;

@Slf4j
public class FirstDemo {

    private static List<Student> students = new ArrayList<>();

    static {
        students.add(new Student("1", "s1", 10, "男"));
        students.add(new Student("2", "s2", 20, "男"));
        students.add(new Student("3", "s3", 30, "男"));
        students.add(new Student("4", "s4", 40, "女"));
        students.add(new Student("5", "s5", 50, "男"));
        students.add(new Student("6", "s6", 60, "男"));
        students.add(new Student("7", "s7", 70, "女"));
        students.add(new Student("8", "s8", 80, "女"));
        students.add(new Student("9", "s9", 40, "男"));
        students.add(new Student("10", "s10", 30, "男"));
        students.add(new Student("11", "s11", 40, "女"));
        students.add(new Student("12", "s12", 10, "男"));
    }


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        log.info("students[{}]", students);
        env.setParallelism(1);
        DataStreamSource<Student> dataStreamSource = env.fromCollection(students, TypeInformation.of(Student.class));


        //创建table 的执行环境
        EnvironmentSettings settings =
                EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);
        Table apiTable = tableEnv.fromDataStream(dataStreamSource).select($("id"),$("name"),
                $("age"), $("gender"))
                ;

        //将table 转换为DateStream
        DataStream<Student> rowDataStream = tableEnv.toAppendStream(apiTable, Student.class);
        rowDataStream.print("Table api");

        //创建视图，用于sql 执行
        tableEnv.createTemporaryView("student",dataStreamSource);
        String querySql = "select * from  student where id >5";
        Table sqlTable = tableEnv.sqlQuery(querySql);
        tableEnv.toAppendStream(sqlTable, Row.class).print("sql");


        env.execute();


    }


}
